package com.study.flume.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 自定义sink，需要继承Flume提供的AbstractSink，实现Configurable接口
 */
public class MySink extends AbstractSink implements Configurable{
    Logger logger = LoggerFactory.getLogger(MySink.class);
    /**
     * 核心处理方法
     *
     * 在Flume处理流程中循环调用
     * @return
     * @throws EventDeliveryException
     */
    @Override
    public Status process() throws EventDeliveryException {
        //休眠一秒钟
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        Status status = null;
        // Start transaction
        //获取channel
        Channel ch = getChannel();
        //获取事务对象
        Transaction txn = ch.getTransaction();
        //开启事务
        txn.begin();
        try {
            //从channel中获取event
            // This try clause includes whatever Channel operations you want to do
            Event event = ch.take();
            // 处理event
             storeSomeData(event);
            //处理成功提交事务
            txn.commit();
            status = Status.READY;
        } catch (Throwable t) {
            //处理失败，回滚事务
            txn.rollback();
            status = Status.BACKOFF;
        }finally {
            //不论事务成功与否，都要关闭
            txn.close();
        }
        return status;
    }

    private void storeSomeData(Event event) {
        String printData= event.getHeaders()+" :: "+new String(event.getBody(), StandardCharsets.UTF_8);
        logger.info(printData);
    }

    @Override
    public void configure(Context context) {

    }
}
